package cn.doit.sparkcore2json

import cn.doit.config.JedisConfig
import org.apache.commons.lang.StringUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext


/**
  * Created by ZHAOXUHUA 
  * .           on 2018/12/1.
  */
object MediaAnaly2Redis {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName(this.getClass.getSimpleName)
    val sc = new SparkContext(conf)
    //操作DataFrame必须有一个SQLContext的实例
    val sqlContext = new SQLContext(sc)
    val file = sqlContext.read.parquet("D:/学习专用/项目/项目四资料/data/dophinData")
    val file1 = sc.textFile("appname")
    val fileData=file.mapPartitions(iter => {
      val jedis = JedisConfig.apply("cls01")

      val result = iter.map(t => {

        //      9 requestmode: Int, 数据请求方式（1:请求、2:展示、3:点击）
        //      36 processnode: Int, 流程节点（1：请求量 kpi 2：有效请求 3：广告请
        //      31 iseffective: Int, 有效标识（有效指可以正常计费的）(0：无效 1：
        //      32 isbilling: Int, 是否收费（0：未收费 1：已收费）
        //      40 isbid: Int, 是否 rtb
        //      43 iswin: Int, 是否竞价成功
        //      42 winprice: Double, rtb 竞价成功价格
        //      76 adpayment: Double, 转换后的广告消费
        //      3 adorderid: Int, 广告 id
        //      15 appname: String, 应用名称
        //14 appid: String, 应用 id
        //      12 requestdate: String, 请求时间,格式为：yyyy-m-dd hh:mm:ss
        //      substring(requestdate, 0, 10)
        // val value = t.get(0)
        val requestmode = t.getAs[Int](8)
        val processnode = t.getAs[Int](35)
        val iseffective = t.getAs[Int](30)
        val isbilling = t.getAs[Int](31)
        val isbid = t.getAs[Int](39)
        val iswin = t.getAs[Int](42)
        val adorderid = t.getAs[Int](2)
        //分割出天
        val day = t.getAs[String](11).substring(0, 10)
        //appname
        var appname = t.getAs[String](14)
        val appid = t.getAs[String](13)
        if (appname == null) {
          if (appid != null) {
            val appNameRedis = jedis.hget("appdict", appid)
            if (StringUtils.isEmpty(appNameRedis)) {
              appname = appNameRedis
            } else {
              appname = appid
            }
          } else {
            appname = "未知"
          }
        }

        //原始请求
        val originalReq = if (1.equals(requestmode) && processnode >= 1) {
          1
        } else {
          0
        }
        //有效请求validReq
        val validReq = if (1.equals(requestmode) && processnode >= 2) {
          1
        } else {
          0
        }
        //广告请求advReq
        val advReq = if (1.equals(requestmode) && processnode == 3) {
          1
        } else {
          0
        }
        //参与竞价partiBid
        val partiBid = if (1.equals(iseffective) && 1.equals(isbilling) && 1.equals(isbid) && !0.equals(adorderid)) {
          1
        } else {
          0
        }
        //竞价成功succBid
        val succBid = if (1.equals(iseffective) && 1.equals(isbilling) && 1.equals(iswin)) {
          1
        } else {
          0
        }
        //展示showT
        val showT = if (2.equals(requestmode) && 1.equals(iseffective)) {
          1
        } else {
          0
        }
        //点击cliT
        val cliT = if (3.equals(requestmode) && 1.equals(iseffective)) {
          1
        } else {
          0
        }
        //广告消费dspAdvConsume
        var dspAdvConsume = if (1.equals(iseffective) && 1.equals(isbilling) && 1.equals(isbid)) {
          t.getAs[Double](41) / 1000

        } else {
          0d
        }
        //广告成本dspAdvCost
        var dspAdvCost = if (1.equals(iseffective) && 1.equals(isbilling) && 1.equals(isbid)) {
          t.getAs[Double](75) / 1000
        } else {
          0d
        }

        ((day, appname), (originalReq, validReq, advReq, partiBid, succBid, showT, cliT, dspAdvConsume, dspAdvCost))
      })
      result
    })

    val redisData: RDD[((String, String), (Int, Int, Int, Int, Int, Int, Int, Double, Double))] = fileData.reduceByKey((t1, t2) => (
      t1._1 + t2._1,
      t1._2 + t2._2,
      t1._3 + t2._3,
      t1._4 + t2._4,
      t1._5 + t2._5,
      t1._6 + t2._6,
      t1._7 + t2._7,
      t1._8 + t2._8,
      t1._9 + t2._9
    ))
    redisData.foreachPartition(it => {
      try {
        val jedis = JedisConfig.apply("cls01")
        it.foreach {
          case ((day, appname), (originalReq, validReq, advReq, partiBid,
          succBid, showT, cliT, dspAdvConsume, dspAdvCost)) => {
            jedis.hset("t_dolphin_Media1", (day, appname).toString,
              (originalReq, validReq, advReq, partiBid, succBid,
                showT, cliT, dspAdvConsume, dspAdvCost).toString())
          }
        }
        jedis.close()
      } catch {
        case e: Exception => {
          e.printStackTrace()
        }
      }
    })
  }
}
